package org.cfk.delaytasktutorial.service.impl;

import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.cfk.delaytasktutorial.enums.LazyJobLazyLevel;
import org.cfk.delaytasktutorial.handler.LazyJobHandler;
import org.cfk.delaytasktutorial.model.DelayedMessage;
import org.cfk.delaytasktutorial.service.LazyJobDeadLetterService;

import org.cfk.delaytasktutorial.service.LazyJobService;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.context.annotation.Lazy;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;

@Service
@Slf4j
public class DelayQueueManager {
    @Resource
    private RedissonClient redissonClient;

    @Resource
    private LazyJobHandler lazyJobHandler;

    @Resource
    @Lazy
    private LazyJobDeadLetterService lazyJobDeadLetterService;

    @Resource
    @Lazy
    private LazyJobService lazyJobService;

    private static final long FIVE_MINUTES = 5 * 60 * 1000;

    private static SimpleAsyncTaskExecutor taskExecutor;

    private volatile boolean isRunning = true;

    private RDelayedQueue<DelayedMessage> delayedQueue;

    static {
        taskExecutor = new SimpleAsyncTaskExecutor("Delay Queue");
        taskExecutor.setConcurrencyLimit(20);
    }

    @PostConstruct
    public void init() {
        RBlockingDeque<DelayedMessage> blockingDeque = redissonClient.getBlockingDeque("delay_queue");
        this.delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
        taskExecutor.submit(this::processDelayedMessages);
    }

    public void addTask(String worker, String content, LocalDateTime executeTime) {
        long delayTime = ChronoUnit.MILLIS.between(LocalDateTime.now(), executeTime);
        if (delayTime < 0) {
            delayTime = 0;
        }
        DelayedMessage message = new DelayedMessage()
                .setType(worker)
                .setBody(content)
                .setDelayTime(delayTime);

         // 5分钟内的任务直接放入Redisson队列
         delayedQueue.offer(message, delayTime, TimeUnit.MILLISECONDS);
         log.info("短延迟任务已加入Redisson队列, worker: {}, delay: {}ms", worker, delayTime);

    }

    private void processDelayedMessages() {
        RBlockingDeque<DelayedMessage> blockingDeque = redissonClient.getBlockingDeque("delay_queue");
        while (isRunning) {
            try {
                List<DelayedMessage> messages = new ArrayList<>();
                DelayedMessage message = blockingDeque.poll(30, TimeUnit.SECONDS);
                if (message != null) {
                    messages.add(message);
                    blockingDeque.drainTo(messages, 99);
                }

                for (DelayedMessage msg : messages) {
                    try {
                        lazyJobHandler.execute(msg.getType(), msg.getBody());
                        log.info("执行延迟任务成功: {}", msg.getType());
                    } catch (Exception e) {
                        log.error("延迟消息处理错误: {}", e.getMessage());
                        
                        // 获取当前重试级别
                        LazyJobLazyLevel currentLevel = msg.getRetryLevel() != null ? 
                            LazyJobLazyLevel.fromCode(msg.getRetryLevel()) : LazyJobLazyLevel.MINUTE10;
                        
                        if (currentLevel == null) {
                            log.error("未找到当前延迟级别: {}", msg.getRetryLevel());
                            continue;
                        }

                        Integer nextLevelCode = currentLevel.getNextLevel();
                        if (nextLevelCode == -1) {
                            // 已达到最大重试次数，加入死信队列
                            lazyJobDeadLetterService.add(msg.getType(), msg.getBody(), LocalDateTime.now());
                            log.info("任务已达到最大重试次数，加入死信队列: {}", msg.getType());
                        } else {
                            // 创建下一级延迟任务
                            LazyJobLazyLevel nextLevel = LazyJobLazyLevel.fromCode(nextLevelCode);
                            if (nextLevel != null) {
                                LocalDateTime nextExecuteTime = LocalDateTime.now().plus(nextLevel.getCode(), ChronoUnit.MINUTES);
                                lazyJobService.add(msg.getType(), msg.getBody(), nextExecuteTime, nextLevel);
                                log.info("创建下一级延迟任务成功: {}, 下次执行时间: {}", msg.getType(), nextExecuteTime);
                            }
                        }
                    }
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            } catch (Exception e) {
                log.error("处理延迟消息时发生错误: {}", e.getMessage());
                try {
                    TimeUnit.SECONDS.sleep(10);
                } catch (InterruptedException ex) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }
    }

    public void shutdown() {
        isRunning = false;
    }
}